package com.atguigu.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by Smexy on 2023/3/5
 *
 * SQL中使用group window
 */
public class Demo10_GroupWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        // 从文件中读取，就是batch，只支持eventtime
        String createTableSql = " create table t1 ( id string, ts bigint  , vc int , " +
            "                   et  as TO_TIMESTAMP_LTZ(ts,3)    ,   " +
            "                   pt  as  PROCTIME() ," +
            "                   WATERMARK FOR et AS et - INTERVAL '1' SECOND   )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";


        // 从kafka读取，就是stream，支持eventtime和processingTime
        String createTableSql2 = " create table t1 ( id string,ts bigint, vc int , " +
            "                   et  as TO_TIMESTAMP_LTZ(ts,3)    ,   " +
            "                   pt  as  PROCTIME() ," +
            "                   WATERMARK FOR et AS et - INTERVAL '1' SECOND   )" +
            " with ( " +
            " 'connector' = 'kafka' ,   " +
            " 'topic' =  'topicC'  ," +
            " 'properties.bootstrap.servers' = 'hadoop102:9092'," +
            "  'properties.group.id' = 'haha', " +
            "  'scan.startup.mode' = 'earliest-offset' ,   " +
            "  'format' = 'json' " +
            "      )                 ";

        tableEnvironment.executeSql(createTableSql);


        /*
            定义窗口 SQL中不支持基于个数的窗口

            TUMBLE(time_attr, interval size)： 定义滚动窗口
            HOP(time_attr, interval slide, interval size)： 定义滑动
            SESSION(time_attr, interval gap)： 定义会话
                    time_attr： 时间字段。区分是处理时间还是事件时间
                    interval ： 格式:  INTERVAL '1' DAY
         */
        String sql1 = " select  id,sum(vc) sumVc ," +
                      " TUMBLE_START(pt, INTERVAL '5' SECONDS) wstart ," +
                      " TUMBLE_END(pt, INTERVAL '5' SECONDS) wend " +
                      "  from t1 " +
                    "   group by TUMBLE(pt, INTERVAL '5' SECONDS),id ";

        String sql2 = " select  id,sum(vc) sumVc ," +
            " HOP_START(et, INTERVAL '3' SECONDS , INTERVAL '5' SECONDS) wstart ," +
            " HOP_END(et, INTERVAL '3' SECONDS , INTERVAL '5' SECONDS) wend " +
            "  from t1 " +
            "   group by HOP(et, INTERVAL '3' SECONDS , INTERVAL '5' SECONDS),id ";


        String sql3 = " select  id,sum(vc) sumVc ," +
            " SESSION_START(et, INTERVAL '3' SECONDS ) wstart ," +
            " SESSION_END(et, INTERVAL '3' SECONDS ) wend " +
            "  from t1 " +
            "   group by SESSION(et, INTERVAL '3' SECONDS ),id ";



        //执行查询
        tableEnvironment.sqlQuery(sql3)
                        .execute()
                        .print();



    }
}
